<?php

namespace util;

use \Kafka;


/*
 * Kafka Helper
 *
 * @author Kevin - xuhuahai@dangdang.com
 * @version 1.0
 * @date Apr 18, 2015
 */
class KafkaClient {
	
	/**
	 *
	 * @var array Array of saved caches for reusing
	 */
	protected static $instances = array ();
	
	/**
	 * 根据缓存名获取Kafka对象
	 * @param unknown $name
	 * @return Ambigous <NULL, multitype:, \Kafka>
	 */
	public static function getKafkaByName($name) {
		global $global_kafka_config;
		soa_debug ( 'getKafkaByConfig : ' . $name . '=' . var_export ( $global_kafka_config [$name], TRUE ) );
		return KafkaClient::getKafkaByConfig ( $global_kafka_config [$name] );
	}
	
	/**
	 * 根据配置获取Kafka对象
	 * @param string $group
	 * @return NULL|multitype:|\Redis
	 */
	public static function getKafkaByConfig($kafka_addrs = false) {
		if (! $kafka_addrs) {
			return NULL;
		}
		
		// ID for database based on the group information
		$id = $kafka_addrs;
		
		// Checking if the same
		if (isset ( self::$instances [$id] )) {
			return self::$instances [$id];
		}
		
		try {
			$instance = new Kafka ($kafka_addrs);
			self::$instances [$id] = $instance;
			return $instance;
		} catch ( PDOException $e ) {
			soa_error ( "getKafkaByConfig error" );
		}
	}
	
	/**
	 * 发布消息
	 * @param unknown $kafka
	 * @param unknown $topic
	 * @param unknown $message
	 */
	public static function produceMessage($kafka, $topic, $message){
		$kafka->produce($topic, $message);
	}
	
	/**
	 * 消费消息
	 * @param unknown $kafka
	 * @param unknown $topic
	 * @param unknown $count
	 * @return unknown
	 */
	public static function consumeMessage($kafka, $topic, $count = -1){
		$msgs = $kafka->consume($topic, Kafka::OFFSET_BEGIN, $count);
		return $msgs;
	}
		
}

?>
